package kafka

import (
	"context"
	"fmt"

	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/Shopify/sarama"
	"github.com/opentracing/opentracing-go"
	"github.com/opentracing/opentracing-go/ext"
	opentracingLog "github.com/opentracing/opentracing-go/log"
)

type Producer struct {
	*ProducerConfig
	sarama.AsyncProducer
	sarama.SyncProducer
}

func (p *Producer) Input(ctx context.Context, msg *sarama.ProducerMessage) (err error) {
	msg.Metadata = ctx
	if !p.Sync {
		p.AsyncProducer.Input() <- msg
	} else {
		if _, _, err = p.SyncProducer.SendMessage(msg); err != nil {
			log.Errorf("input error: %s", err.Error())
		}
	}
	return
}

func (p *Producer) Close() (err error) {
	if !p.Sync {
		if p.AsyncProducer != nil {
			return p.AsyncProducer.Close()
		}
	}
	if p.SyncProducer != nil {
		return p.SyncProducer.Close()
	}
	return
}

func (p *Producer) errProcess() {
	err := p.Errors()
	for {
		e, ok := <-err
		if !ok {
			return
		}
		key, _ := e.Msg.Key.Encode()
		log.Errorf("kafka producer fault to send message, topic[%s], partition[%s], offset[%d], key[%s]",
			e.Msg.Topic, e.Msg.Partition, e.Msg.Offset, string(key), e.Err.Error(),
		)
		if p.EnableTrace {
			span, _ := opentracing.StartSpanFromContext(e.Msg.Metadata.(context.Context),
				"kafka.producer."+e.Msg.Topic, ext.SpanKindProducer)
			ext.Component.Set(span, "golang-kafka")
			ext.PeerService.Set(span, "kafka")
			ext.MessageBusDestination.Set(span, e.Msg.Topic)
			headers := make(map[string]string)
			_ = opentracing.GlobalTracer().Inject(
				span.Context(),
				opentracing.TextMap,
				opentracing.TextMapCarrier(headers),
			)
			for key, value := range headers {
				header := sarama.RecordHeader{
					Key:   []byte(key),
					Value: []byte(value),
				}
				e.Msg.Headers = append(e.Msg.Headers, header)
			}
			span.LogFields(
				opentracingLog.String("event", "error"),
				opentracingLog.String("message", fmt.Sprintf("%+v", e.Err)))
			span.Finish()
		}
		if p.errHandle != nil {
			p.errHandle(e)
		}
	}
}

func (p *Producer) successProcess() {
	suc := p.Successes()
	for {
		msg, ok := <-suc
		if !ok {
			return
		}
		key, _ := msg.Key.Encode()
		log.Debugf("kafka producer send message success, topic[%s], partition[%s], offset[%d], key[%s]",
			msg.Topic, msg.Partition, msg.Offset, string(key),
		)
		if p.EnableTrace {
			span, _ := opentracing.StartSpanFromContext(msg.Metadata.(context.Context),
				"kafka.producer."+msg.Topic, ext.SpanKindProducer)
			ext.Component.Set(span, "golang-kafka")
			ext.PeerService.Set(span, "kafka")
			ext.MessageBusDestination.Set(span, msg.Topic)
			headers := make(map[string]string)
			_ = opentracing.GlobalTracer().Inject(
				span.Context(),
				opentracing.TextMap,
				opentracing.TextMapCarrier(headers),
			)
			for key, value := range headers {
				header := sarama.RecordHeader{
					Key:   []byte(key),
					Value: []byte(value),
				}
				msg.Headers = append(msg.Headers, header)
			}
			span.Finish()
		}
		if p.sucHandle != nil {
			p.sucHandle(msg)
		}
	}
}
